package com.atguigu.userprofile.app

import com.atguigu.userprofile.common.bean.TagInfo
import com.atguigu.userprofile.common.constants.ConstCode
import com.atguigu.userprofile.common.dao.TagInfoDAO
import com.atguigu.userprofile.common.util.MyClickhouseUtil
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

object TaskBitmapApp {

  //读取：  clickhouse 宽表
  //写入 ：  clickhouse bitmap表

  def main(args: Array[String]): Unit = {

    //为了方便调度监控管理程序 即使没有使用分布式计算 依然使用spark程序运行
    // driver  executor
    //如果 不加以下两句 yarn会报错 未找到spark context
    val sparkConf: SparkConf = new SparkConf().setAppName("task_bitmap_app")//.setMaster("local[*]")
    val sparkContext = new SparkContext(sparkConf)


    //1  读取：  clickhouse 宽表
    //select ttg.tg.1 as tag_code,ttg.tg.2 as tag_value,groupBitmapState(uid)  uid_bit
    //from
    //(
    //select uid,  arrayJoin( [('gender',gender),('agegroup',agegroup),('favor',favor)])  tg from user_tag_merge
    //)ttg
    //group by ttg.tg.1,ttg.tg.2
    val taskDate: String = args(1)
    val tableName = s"user_tag_merge_${taskDate.replace("-", "")}"

    val tagInfoList: List[TagInfo] = TagInfoDAO.getTagInfoWithOn()
    // 因为要 根据标签值类型 分别插入到4个bitmap表中  所以要把标签集合拆成四份 分别组合sql插入
    val tagInfoListString = new ListBuffer[TagInfo]
    val tagInfoListLong = new ListBuffer[TagInfo]
    val tagInfoListDecimal = new ListBuffer[TagInfo]
    val tagInfoListDate = new ListBuffer[TagInfo]
    for (tagInfo <- tagInfoList) {
      if (tagInfo.tagValueType == ConstCode.TAG_VALUE_TYPE_STRING) {
        tagInfoListString.append(tagInfo)
      } else if (tagInfo.tagValueType == ConstCode.TAG_VALUE_TYPE_LONG) {
        tagInfoListLong.append(tagInfo)
      } else if (tagInfo.tagValueType == ConstCode.TAG_VALUE_TYPE_DECIMAL) {
        tagInfoListDecimal.append(tagInfo)
      } else if (tagInfo.tagValueType == ConstCode.TAG_VALUE_TYPE_DATE) {
        tagInfoListDate.append(tagInfo)
      }
    }
    //根据不同的tagInfoList 插入到对应的bitmap表中
    insertBitmap(tagInfoListString.toList, tableName, "user_tag_value_string", taskDate)
    insertBitmap(tagInfoListLong.toList, tableName, "user_tag_value_long", taskDate)
    insertBitmap(tagInfoListDecimal.toList, tableName, "user_tag_value_decimal", taskDate)
    insertBitmap(tagInfoListDate.toList, tableName, "user_tag_value_date", taskDate)

  }

  //根据不同的tagInfoList 插入到对应的bitmap表中
  def insertBitmap(tagInfoList: List[TagInfo], sourceTableName: String, targetTableName: String, taskDate: String): Unit = {
    if (tagInfoList.size > 0) {
      val tagCodeList: List[String] = tagInfoList.map(tagInfo => s"('${tagInfo.tagCode}',${tagInfo.tagCode.toLowerCase()})")
      val tagCodeSQL = tagCodeList.mkString(",")

      //保证数据幂等性  每次插入数据前 清空分区
      val  deleteSQL = s" alter table $targetTableName delete where dt ='$taskDate'"
      println(deleteSQL)
      MyClickhouseUtil.executeSql(deleteSQL)


      val selectSQL =
        s"""
           |select ttg.tg.1 as tag_code,ttg.tg.2 as tag_value,groupBitmapState( cast( uid as UInt64))  uid_bit,
           |'$taskDate'
           |from
           |(
           |select uid,  arrayJoin( [${tagCodeSQL}])  tg from  ${sourceTableName}
           |)ttg
           |where ttg.tg.2<>''
           |group by ttg.tg.1,ttg.tg.2
         |
       """.stripMargin
      //写入 ：  clickhouse bitmap表
      val insertSQL = s" insert into $targetTableName $selectSQL"
      println(insertSQL)

      MyClickhouseUtil.executeSql(insertSQL)

    }
  }

}
